package flink_p1

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 *
 *
 * 每1分钟 对最近1分钟的车速进行排序 输出max min 并将排序结果发给下游
 * 分析：需要全量数据，=> 需要全量聚合函数
 * */

object FlinkTest_18_window_process {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)

    socketStream
      .map(data => {
        val arr: Array[String] = data.split(" ")
        (arr(0), arr(1).toInt)
      }).timeWindowAll(Time.seconds(20), Time.seconds(20))
      .process(new ProcessAllWindowFunction[(String, Int), List[(String, Int)], TimeWindow] {

        override def process(context: Context, elements: Iterable[(String, Int)], out: Collector[List[(String, Int)]]): Unit = {


          val sorted: List[(String, Int)] = elements.toList.sortBy(_._2)
          out.collect(sorted)
          println(s"max: ${sorted.last}")
          println(s"min: ${sorted.head}")

        }
      }).print()


    env.execute()
  }

}
